package com.sjgs.gis.config;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import com.sjgs.common.utils.KryoUtils;
import com.sjgs.gis.fs.store.DefaultKVStore;
import com.sjgs.common.domain.Node;
import com.sjgs.common.ketama.DefaultHashAlgorithm;
import com.sjgs.common.ketama.KetamaNodeKeyFormatter;
import com.sjgs.common.ketama.KetamaNodeLocator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.springframework.cloud.client.DefaultServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.cloud.loadbalancer.core.*;
import reactor.core.publisher.Mono;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.reactive.DefaultResponse;
import org.springframework.cloud.client.loadbalancer.reactive.EmptyResponse;
import org.springframework.cloud.client.loadbalancer.reactive.Response;

import static com.sjgs.common.Constants.SEP;

/**
 * A Round-Robin-based implementation of {@link ReactorServiceInstanceLoadBalancer}.
 *
 * @author Spencer Gibb
 * @author Olga Maciaszek-Sharma
 */
public class KetamaHashLoadBalancer {

    private static final Log log = LogFactory.getLog(KetamaHashLoadBalancer.class);
    private DiscoveryClient discoveryClient;
    private DefaultKVStore kvstore;
    private final String serviceId;
    private final RocksDB sysdb;

    public KetamaHashLoadBalancer(DiscoveryClient discoveryClient, DefaultKVStore kvstore, String serviceId) {
        this.discoveryClient = discoveryClient;
        this.kvstore = kvstore;
        this.serviceId = serviceId;
        this.sysdb = kvstore.getDb();
    }

    public Mono<Response<ServiceInstance>> choose(URI uri) {
        List<Node> nodes = new ArrayList<>();
        // first get from kvstore cache
        //TODO 定时30分钟监视instances记录，并动态处理一致性哈希导致的数据倾斜
        String search_key = serviceId + SEP + "instances";
        RocksIterator iter = sysdb.newIterator(kvstore.getDefaultColumnFamilyHandle());
        for (iter.seek(search_key.getBytes()); iter.isValid(); iter.next()) {
            if (new String(iter.key()).equals(search_key)) {
                nodes = KryoUtils.deserializeList(iter.value());
            }
        }
        // no cache instances
        if (nodes.size() == 0) {
            // second get from discoveryClient and save to kvstore
            nodes = getInstances(serviceId);
        }
        return Mono.just(getInstanceResponse(nodes, uri));
    }

    //TODO 定期查询更新kvstore instances表，发生变化触发数据迁移计划
    public List<Node> getInstances(String serviceId) {
        List<Node> discovery_nodes = new ArrayList<>();
        if (discoveryClient != null) {
            String search_key = serviceId + SEP + "instances";
            discovery_nodes = discoveryClient.getInstances(this.serviceId)
                    .stream()
                    .map(it -> new Node(it.getInstanceId(), serviceId, it.getHost(), it.getPort()))
                    .collect(Collectors.toList());
            //cache node
            try {
                sysdb.put(kvstore.getDefaultColumnFamilyHandle(), search_key.getBytes(), KryoUtils.serializeList(discovery_nodes));
            } catch (RocksDBException e) {
                ;
            }
        }
        return discovery_nodes;
    }

    public Response<ServiceInstance> getInstanceResponse(List<Node> nodes, URI uri) {
        if (nodes.size() > 0) {
            //负载权重
            Map<InetSocketAddress, Integer> weights = new HashMap<>();
            //负载定位器
            KetamaNodeLocator ketamaNodeLocator = new KetamaNodeLocator(nodes,
                    DefaultHashAlgorithm.KETAMA_HASH, KetamaNodeKeyFormatter.Format.INSTANCEID,
                    weights);
            //获取数据hash
            String key = uri.getPath().substring(uri.getPath().lastIndexOf("/") + 1, uri.getPath().length());
            //定位主节点
            Node primaryNode = ketamaNodeLocator.getPrimary(key);
            System.out.println(key + ":" + primaryNode.getInstanceId());
            Map<String, String> metadata = new HashMap<>();
            return new DefaultResponse(new DefaultServiceInstance(primaryNode.getInstanceId(), primaryNode.getServiceId(), primaryNode.getHost(), primaryNode.getPort(), false, metadata));
        } else {
            return new EmptyResponse();
        }
    }
}